package com.feiwo.nonego.rpc.provider.common.handler;

import com.feiwo.nonego.rpc.common.threadpool.ServerThreadPool;
import com.feiwo.nonego.rpc.common.util.RpcServiceUtils;
import com.feiwo.nonego.rpc.protocol.RpcProtocol;
import com.feiwo.nonego.rpc.protocol.enums.RpcStatus;
import com.feiwo.nonego.rpc.protocol.enums.RpcType;
import com.feiwo.nonego.rpc.protocol.request.RpcRequest;
import com.feiwo.nonego.rpc.protocol.response.RpcResponse;
import com.feiwo.nonego.rpc.reflect.api.ReflectInvoker;
import com.feiwo.nonego.rpc.spi.loader.ExtensionLoader;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.Objects;

/**
 * @author feiwo 2025-08-04
 */
public class RpcProviderHandler extends SimpleChannelInboundHandler<RpcProtocol<RpcRequest>> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcProviderHandler.class);

    private final Map<String,Object> handlerMap;
    private final ReflectInvoker reflectInvoker;

    public RpcProviderHandler(String reflectType, Map<String, Object> handlerMap) {
        this.reflectInvoker = ExtensionLoader.getExtension(ReflectInvoker.class, reflectType);
        this.handlerMap = handlerMap;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcProtocol<RpcRequest> msg) throws Exception {
        ServerThreadPool.submit(() -> {
            var header = msg.getHeader();
            var request = msg.getBody();
            header.setMsgType((byte) RpcType.RESPONSE.getType());
            var responseProtocol = new RpcProtocol<RpcResponse>();
            var response = new RpcResponse();

            try {
                var result = handle(request);
                response.setResult(result);
                response.setAsync(request.isAsync());
                response.setOneway(request.isOneway());
                header.setStatus((byte) RpcStatus.SUCCESS.getCode());
            } catch (Throwable e) {
                response.setError(e.getMessage());
                header.setStatus((byte) RpcStatus.FAIL.getCode());
                LOGGER.error("RPC Server handle request error", e);
            }
            responseProtocol.setHeader(header);
            responseProtocol.setBody(response);
            ctx.writeAndFlush(responseProtocol).addListener((ChannelFutureListener) future -> {
                LOGGER.debug("Send Response for request ID: {}", header.getMsgId());
            });
        });
    }

    /**
     * 处理请求
     * @param request 请求
     * @return 响应
     * @throws Throwable 抛出的异常
     */
    private Object handle(RpcRequest request) throws Throwable {
        var serviceKey = RpcServiceUtils.buildServiceKey(request.getClassName(), request.getVersion(), request.getGroup());
        var serviceBean = handlerMap.get(serviceKey);
        if (Objects.isNull(serviceBean)) {
            throw new RuntimeException("未找到服务：" + serviceKey);
        }

        var serviceClass = serviceBean.getClass();
        var methodName = request.getMethodName();
        var parameterTypes = request.getParameterTypes();
        var parameters = request.getParameters();

        LOGGER.debug("serviceName: {}, methodName: {}, parameterType: {}, parameters: {}", serviceClass.getName(), methodName, parameterTypes, parameters);

        return this.reflectInvoker.invoke(serviceBean, serviceClass, methodName, parameterTypes, parameters);
    }
}
